package com.hhf.rrd.wordcount;

import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * WordCount 流式任务
 *      开启socket端口：nc -l 9999
 *      查看端口占用: lsof -i :9999
 *
 * @author huanghaifeng15
 * @date 2022/2/9 20:30
 **/
public class WordCountStreamApp {

    public static void main(String[] args) throws Exception {
        // 获取上下文环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 获取流式数据源
        DataStreamSource<String> source = env.socketTextStream("localhost", 9999);

        // 处理词频的业务逻辑  Transformation
        SingleOutputStreamOperator<Tuple2<String, Integer>> operator = source.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                if (StringUtils.isNotBlank(value)) {
                    String[] wordArr = value.split(",");
                    for (String word : wordArr) {
                        out.collect(word.trim());
                    }
                }
            }
        }).map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                return new Tuple2<>(value, 1);
            }
        }).keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            @Override
            public String getKey(Tuple2<String, Integer> value) throws Exception {
                return value.f0;
            }
        }).sum(1);

        // sink 处理
        operator.print();

        // 流式任务，需要启动服务程序，保持任务对于流式数据的持续性监听
        env.execute();
    }
}
